/*
 * ReplicaManagerResolver.java    0.0.1    26/dic/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transparency.impl;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.mentalsmash.tazio.commons.identifiers.NID;
import org.mentalsmash.tazio.net.cs.client.Connection;
import org.mentalsmash.tazio.net.cs.client.RemoteServer;
import org.mentalsmash.tazio.net.cs.client.SocketRemoteServer;
import org.mentalsmash.tazio.transparency.commons.messages.BeginConnectionMessage;
import org.mentalsmash.tazio.transparency.commons.messages.ConnectionEstablished;
import org.mentalsmash.tazio.transparency.commons.messages.DisconnectionMessage;
import org.mentalsmash.tazio.transparency.interfaces.DeliveryFailureListener;
import org.mentalsmash.tazio.transparency.interfaces.Dispatcher;
import org.mentalsmash.tazio.transparency.interfaces.InputMessageListener;
import org.mentalsmash.tazio.transparency.interfaces.Resolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.internal.Preconditions;
import com.google.inject.name.Named;

/**
 * This is the base server side implementation of the {@link Resolver} and
 * {@link InputMessageListener} interfaces, the latter one used to manage
 * {@link BeginConnectionMessage}s.
 * <p>
 * This implementation differs from the WeakReplicaManager's one because it has
 * to handle the association between {@link TazioNodeID}s and
 * {@link RemoteServer}s, with a different {@link RemoteServer} for each
 * WeakReplicaManager connected to it. This approach is different from the one
 * used in the WeakReplicaManager resolver since it had to manage identifiers
 * and addresses, keeping only one {@link RemoteServer} connected.
 * </p>
 * <p>
 * Furthermore, this class behaves as an {@link InputMessageListener} for
 * {@link BeginConnectionMessage}s, in order to establish a new connection with
 * the clients.
 * </p>
 * 
 * @version 0.0.1 26/dic/2009
 * @author Enrico Savioli <savioli.enrico@gmail.com>
 */
@Singleton
public class ReplicaManagerResolver implements Resolver, InputMessageListener,
	DeliveryFailureListener {

    private final static Logger log = LoggerFactory
	    .getLogger(ReplicaManagerResolver.class);

    private final Dispatcher dispatcher;

    private final Map<NID, RemoteServer> connectedClients;
    private final int POOL_SIZE;
    private final int MAX_POOL_SIZE;
    private final long KEEP_ALIVE_TIME;
    private final int MAX_CONNECTION_REPETITIONS;

    private final LinkedBlockingQueue<Runnable> queue;
    private final ThreadPoolExecutor executor;

    @Inject
    public ReplicaManagerResolver(Dispatcher dispatcher,
	    @Named("pool_size") int POOL_SIZE,
	    @Named("max_pool_size") int MAX_POOL_SIZE,
	    @Named("keep_alive_time") long KEEP_ALIVE_TIME,
	    @Named("max_connection_repetitions") int MAX_CONNECTION_REPETITIONS) {

	log.debug("Initializing the Resolver.");

	this.dispatcher = dispatcher;
	this.dispatcher.register(BeginConnectionMessage.class, this);
	
	//listening for disconnections too..
	this.dispatcher.register(DisconnectionMessage.class, new DisconnectionMessageHandler(this));

	this.POOL_SIZE = POOL_SIZE;
	this.MAX_POOL_SIZE = MAX_POOL_SIZE;
	this.KEEP_ALIVE_TIME = KEEP_ALIVE_TIME;
	this.MAX_CONNECTION_REPETITIONS = MAX_CONNECTION_REPETITIONS;

	this.connectedClients = new HashMap<NID, RemoteServer>();

	queue = new LinkedBlockingQueue<Runnable>();
	executor = new ThreadPoolExecutor(this.POOL_SIZE, this.MAX_POOL_SIZE,
		this.KEEP_ALIVE_TIME, TimeUnit.SECONDS, queue);

	log.debug("Resolver Initialized.");
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.interfaces.Resolver#getAddress(org
     * .mentalsmash.tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public InetAddress getAddress(NID nid) {

	synchronized (connectedClients) {

	    RemoteServer remoteServer = null;

	    remoteServer = connectedClients.get(nid);

	    if (remoteServer != null) {

		return remoteServer.getAddress();

	    }

	}

	return null;
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.mentalsmash.tazio.transparency.interfaces.Resolver#getNIDList()
     */
    @Override
    public List<NID> getNIDList() {

	synchronized (connectedClients) {

	    return new ArrayList<NID>(connectedClients.keySet());

	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.interfaces.Resolver#registerMapping
     * (org.mentalsmash.tazio.commons.identifiers.TazioNodeID,
     * java.net.InetAddress)
     */
    @Override
    public void registerMapping(NID nid, InetAddress address) {

	throw new RuntimeException("Mapping registration not implemented yet!");

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.interfaces.Resolver#removeMapping(
     * org.mentalsmash.tazio.commons.identifiers.TazioNodeID,
     * java.net.InetAddress)
     */
    @Override
    public void removeMapping(NID nid) {

	synchronized (connectedClients) {
	    

	    RemoteServer removed = connectedClients.remove(nid);
	    if(removed == null){
		log.warn("Cannot remove: No mapping found for " + nid);
	    }
	    log.debug("Mapping for {} removed", nid);

	}

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transparency.interfaces.Resolver#resolve(org.
     * mentalsmash.tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public Connection resolve(NID nid) throws TransparentDeliveryException {

	synchronized (connectedClients) {

	    RemoteServer remoteServer = connectedClients.get(nid);
	    Connection connection = remoteServer.getConnection();
	    return connection;

	}

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transparency.interfaces.InputMessageListener#
     * handleMessage(org.mentalsmash.tazio.commons.identifiers.TazioNodeID,
     * java.io.Serializable)
     */
    @Override
    public <T extends Serializable> Serializable handleMessage(NID nid,
	    T message) {
	
	Preconditions.checkNotNull(message);
	Preconditions.checkArgument(message instanceof BeginConnectionMessage);
	
	// System.out.println(message.getClass());

	synchronized (connectedClients) {

	    if (connectedClients.containsKey(nid)) {
		return new ConnectionEstablished();
	    }

	}

	BeginConnectionMessage connectionMessage = (BeginConnectionMessage) message;

	int port = connectionMessage.getPort();
	InetAddress address = connectionMessage.getAddress();

	RemoteServer remoteServer = new SocketRemoteServer(executor, address,
		port);

	for (int i = 0; i < MAX_CONNECTION_REPETITIONS; i++) {

	    try {

		remoteServer.connect();

	    } catch (IOException e) {

		String msg = "Input/Output error while connecting to client's facade.";
		log.error(msg);
		e.printStackTrace();

		if (i == MAX_CONNECTION_REPETITIONS - 1) {

		    return null;

		}

		continue;
	    }

	    // System.out.println("Exiting from the connection cycle of the RM");

	    break;

	}

	synchronized (connectedClients) {

	    if (!connectedClients.containsKey(nid)) {
		connectedClients.put(nid, remoteServer);
	    }

	}

	return new ConnectionEstablished();
    }

    public long getKeepAliveTime() {

	return this.KEEP_ALIVE_TIME;

    }

    public int getPoolSize() {

	return this.POOL_SIZE;
    }

    public int getMaxPoolSize() {

	return this.MAX_POOL_SIZE;

    }

    public int getMaxConnectionRepetitions() {

	return this.MAX_CONNECTION_REPETITIONS;

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.interfaces.DeliveryFailureListener
     * #failedDelivery(org.mentalsmash.tazio.commons.identifiers.TazioNodeID)
     */
    @Override
    public void failedDelivery(NID nid) {
	log.warn("Failed delivering a message to " + nid);
	// TODO delete the RemoteServer object for NID

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transparency.interfaces.Resolver#
     * setLocalListeningAddress(java.net.InetAddress, int)
     */
    @Override
    public void setLocalListeningAddress(InetAddress address, int port) {
	throw new UnsupportedOperationException();

    }
    

}
